package com.codejiwei.flink.cep;

import com.codejiwei.flink.entity.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.nfa.aftermatch.AfterMatchSkipStrategy;
import org.apache.flink.cep.nfa.aftermatch.NoSkipStrategy;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * @author jiwei
 * @description
 * @date 2023/5/26 14:36
 */
public class Flink_CEP_Condition {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(10);

        //读取文件并指定事件时间
//        SingleOutputStreamOperator<WaterSensor> stream = env.readTextFile("src/main/resources/data/sensor-data.log")
        SingleOutputStreamOperator<WaterSensor> stream = env.readTextFile("src/main/resources/data/sensor.txt")
                .map(line -> {
                    String[] fields = line.split(",");
                    return new WaterSensor(fields[0], Long.valueOf(fields[1]), Integer.valueOf(fields[2]));
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner((sensor, recordTimestamp) -> sensor.getTs()));


        Pattern<WaterSensor, WaterSensor> pattern1 = Pattern
                .<WaterSensor>begin("pattern1")
                .where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor waterSensor) throws Exception {
                        System.out.println("=====" + waterSensor);
                        return waterSensor.getId().equals("sensor_1");
                    }
                });



        Pattern<WaterSensor, WaterSensor> pattern2 = Pattern
                .<WaterSensor>begin("pattern2")
                .where(new IterativeCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor waterSensor, Context<WaterSensor> context) throws Exception {
                        long processingTime = context.currentProcessingTime();
                        long timestamp = context.timestamp();
                        System.out.println("processingTime: " + processingTime + ", timestamp: " + timestamp);
                        return waterSensor.getVc() <= 3;
                    }
                });

        Pattern<WaterSensor, WaterSensor> pattern3 = Pattern
                .<WaterSensor>begin("pattern3")
                .where(new IterativeCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor value, Context<WaterSensor> ctx) throws Exception {
                        return "sensor_1".equals(value.getId());
                    }
                })
                .where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor waterSensor) throws Exception {
                        return waterSensor.getVc() > 30;
                    }
                })
                .or(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor waterSensor) throws Exception {
                        return waterSensor.getVc() <= 3;
                    }
                });

        Pattern<WaterSensor, WaterSensor> pattern4 = Pattern
                .<WaterSensor>begin("pattern4")
                .where(new IterativeCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor value, Context<WaterSensor> ctx) throws Exception {
                        return "sensor_1".equals(value.getId());
                    }
                })
                .timesOrMore(3)
                .until(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor waterSensor) throws Exception {
                        return waterSensor.getVc() >= 30;
                    }
                });


        Pattern<WaterSensor, WaterSensor> pattern5 = Pattern
                .<WaterSensor>begin("start")
                .where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor waterSensor) throws Exception {
                        return "sensor_1".equals(waterSensor.getId());
                    }
                })
                .next("end")
                .where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor waterSensor) throws Exception {
                        return "sensor_2".equals(waterSensor.getId());
                    }
                });

        //严格连续模式 通过迭代条件可以获取上一个模式中的数据
        Pattern<WaterSensor, WaterSensor> pattern6 = Pattern
                .<WaterSensor>begin("start")
                .where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor waterSensor) throws Exception {
                        return "sensor_1".equals(waterSensor.getId());
                    }
                })
                .next("end")
//                .notNext("not_next")
                .where(new IterativeCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor value, Context<WaterSensor> ctx) throws Exception {
//                        Iterator<WaterSensor> start = ctx.getEventsForPattern("start").iterator();
//                        while (start.hasNext()) {
//                            System.out.println("====" + start.next());
//                        }
//                        return value.getVc() <= 3;
                        return "sensor_1".equals(value.getId());
                    }
                });

        // next是严格模式 followedBy是松散模式 followedByAny是不确定的松散模式
        Pattern<WaterSensor, WaterSensor> pattern7 = Pattern
                .<WaterSensor>begin("start")
                .where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor waterSensor) throws Exception {
                        return "sensor_1".equals(waterSensor.getId());
                    }
                })
//                .followedBy("end")
//                .next("end")
                .followedByAny("end")
                .where(new IterativeCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor value, Context<WaterSensor> ctx) throws Exception {
                        return "sensor_3".equals(value.getId());
                    }
                });


        //循环模式默认为 松散连续 可以设置为严格连续或者非确定的松散连续
        Pattern<WaterSensor, WaterSensor> pattern8 = Pattern
                .<WaterSensor>begin("start")
                .where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor waterSensor) throws Exception {
                        return "sensor_1".equals(waterSensor.getId());
                    }
                })
                .times(3)
                .allowCombinations();
//                .consecutive();


        //循环模式的贪婪性
        Pattern<WaterSensor, WaterSensor> pattern9 = Pattern
                .<WaterSensor>begin("start")
                .where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor waterSensor) throws Exception {
                        return "sensor_1".equals(waterSensor.getId());
                    }
                })
                .times(2, 3).greedy()
                .next("end")
                .where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor waterSensor) throws Exception {
                        return waterSensor.getVc() == 30;
                    }
                });


        //模式的可选性
        Pattern<WaterSensor, WaterSensor> pattern10 = Pattern
                .<WaterSensor>begin("start")
                .where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor waterSensor) throws Exception {
                        return "sensor_1".equals(waterSensor.getId());
                    }
                })
//                .optional().times(2)
                .times(2).optional()
                .next("end")
                .where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor waterSensor) throws Exception {
                        return "sensor_2".equals(waterSensor.getId());
                    }
                });

        //模式组
        Pattern<WaterSensor, WaterSensor> pattern11 = Pattern
                .begin(Pattern
                        .<WaterSensor>begin("start")
                        .where(new SimpleCondition<WaterSensor>() {
                            @Override
                            public boolean filter(WaterSensor waterSensor) throws Exception {
                                return "sensor_1".equals(waterSensor.getId());
                            }
                        })
                        .next("next")
                        .where(new SimpleCondition<WaterSensor>() {
                            @Override
                            public boolean filter(WaterSensor waterSensor) throws Exception {
                                return "sensor_2".equals(waterSensor.getId());
                            }
                        })
                ).times(2);

        NoSkipStrategy skip = AfterMatchSkipStrategy.noSkip();

        //超时数据
        Pattern<WaterSensor, WaterSensor> pattern12 = Pattern
                .<WaterSensor>begin("start", skip)
                .where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor waterSensor) throws Exception {
//                        Thread.sleep(2000);
                        return "sensor_1".equals(waterSensor.getId());
                    }
                })
                .next("next")
                .where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor waterSensor) throws Exception {
                        return "sensor_2".equals(waterSensor.getId());
                    }
                })
                .within(Time.seconds(1));





        CEP.pattern(stream, pattern12)
                .select(new PatternSelectFunction<WaterSensor, String>() {
                    @Override
                    public String select(Map<String, List<WaterSensor>> pattern) throws Exception {
                        return pattern.toString();
                    }
                })
                .print();


        env.execute();
    }
}
